#pragma once
/*
    对于客户端内容的整合封装
    -业务功能:1.基础rpc功能 2.服务注册功能 3.服务发现功能  4.服务发布与订阅功能
    -基础底层模块:网络通信客户端
*/

#include "../common/dispatcher.hpp"
#include "requestor.hpp"
#include "rpc_caller.hpp"
#include "rpc_registry.hpp"
#include "rpc_topic.hpp"

namespace tmh_rpc
{
    namespace client
    {
        class RegistryClient
        {
        public:
            using ptr = std::shared_ptr<RegistryClient>;
            // 构造函数传入注册中心的地址信息,用于发送连接注册服务请求
            RegistryClient(const string &ip, int port) : _requestor(std::make_shared<Requestor>()),
                                                         _provider(std::make_shared<client::Provider>(_requestor)),
                                                         _dispatcher(std::make_shared<Dispatcher>())
            {
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<BaseMessage>(MType::RSP_SERVICE, rsp_cb);

                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                _client = ClientFactory::create(ip, port);

                _client->setMessageCallback(message_cb);

                // 连接服务器
                _client->connect();
            }

            // 向外提供的服务注册接口
            bool registryMethod(const string &method, const Address &host)
            {
                return _provider->registryMethod(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _requestor;
            client::Provider::ptr _provider;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        class DiscoveryClient
        {
        public:
            using ptr = std::shared_ptr<DiscoveryClient>;
            // 构造函数传入注册中心的地址信息,用于连接注册中心
            DiscoveryClient(const string &ip, int port, const Discoverer::offlineCallback &cb) : _requestor(std::make_shared<Requestor>()),
                                                                                                 _discoverer(std::make_shared<client::Discoverer>(_requestor, cb)),
                                                                                                 _dispatcher(std::make_shared<Dispatcher>())
            {

                // 给dispatch模块设置回调函数
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<BaseMessage>(MType::RSP_SERVICE, rsp_cb); // 注册消息类型和回调的映射关系
                auto req_cb = std::bind(&client::Discoverer::onServiceRequest, _discoverer.get(),
                                        std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<ServiceRequest>(MType::REQ_SERVICE, req_cb); // 注册消息类型和回调的映射关系
                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                _client = ClientFactory::create(ip, port);
                _client->setMessageCallback(message_cb);
                _client->connect();
            }

            // 向外提供的服务发现接口
            bool serviceDiscovery(const string &method, Address &host)
            {
                return _discoverer->serviceDiscovery(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _requestor;
            client::Discoverer::ptr _discoverer;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        // 短连接会存在异步调用的无法接收的问题,因此采用长连接
        class RpcClient
        {
        public:
            using ptr = std::shared_ptr<RpcClient>;

            // enableDiscovery:是否启用服务发现功能,也决定了传入的地址信息是服务注册中心还是服务提供者
            RpcClient(bool enableDiscovery, const string &ip, int port) : _enableDiscovery(enableDiscovery),
                                                                          _requestor(std::make_shared<Requestor>()),
                                                                          _dispatcher(std::make_shared<Dispatcher>()),
                                                                          _caller(std::make_shared<client::RpcCaller>(_requestor))
            {
                // 针对rpc请求后的响应进行的回调处理
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MType::RSP_RPC, rsp_cb);
                // 如果启用了服务发现,地址信息就是注册中心的地址:是服务发现客户端需要连接的地址,则通过地址信息实例化discovery_client
                // 如果没有启用服务发现,则地址信息是服务提供者的地址,则直接实例化好rpc_client
                if (_enableDiscovery)
                {
                    auto offline_cb = std::bind(&RpcClient::delClient, this, std::placeholders::_1);
                    _discovery_client = std::make_shared<DiscoveryClient>(ip, port, offline_cb);
                }
                else
                {
                    // 没有启用服务发现
                    auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                                std::placeholders::_1, std::placeholders::_2);
                    _rpc_client = ClientFactory::create(ip, port);
                    _rpc_client->setMessageCallback(message_cb);
                    _rpc_client->connect();
                }
            }

            // 同步,异步,回调的处理函数
            bool call(const string &method, const Json::Value &params, Json::Value &result)
            {
                // 通过连接向服务提供者发生请求
                // 1. 获取到服务提供者的信息---a.服务发现  b.固定的服务提供者
                BaseClient::ptr client = getClient(method);
                if (client.get() == nullptr)
                    return false; // 说明没有服务提供者

                // 通过客户端连接, 发送rpc请求
                return _caller->call(client->connection(), method, params, result);
            }

            bool call(const string &method, const Json::Value &params, RpcCaller::JsonAsyncResponse &result)
            {
                // 同步调用和异步的调用的不同在于result的类型不同
                BaseClient::ptr client = getClient(method);
                if (client.get() == nullptr)
                    return false; // 说明没有服务提供者

                // 通过客户端连接, 发送rpc请求
                return _caller->call(client->connection(), method, params, result);
            }
            bool call(const string &method, const Json::Value &params, const RpcCaller::JsonResponseCallback &cb)
            {
                BaseClient::ptr client = getClient(method);
                if (client.get() == nullptr)
                    return false; // 说明没有服务提供者

                // 通过客户端连接, 发送rpc请求
                return _caller->call(client->connection(), method, params, cb);
            }

        private:
            // 对_rpc_clients的接口进行封装
            BaseClient::ptr newClient(const Address &host)
            {
                // 创建一个新的客户端,设置其回调函数
                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                auto client = ClientFactory::create(host.first, host.second);
                client->setMessageCallback(message_cb);
                client->connect();
                putClient(host, client);
                return client;
            }
            BaseClient::ptr getClient(const Address &host)
            {
                // 根据传入的地址信息返回客户端信息
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it == _rpc_clients.end())
                {
                    return BaseClient::ptr();
                }
                return it->second;
            }

            void putClient(const Address &host, BaseClient::ptr &client)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // TODO
                _rpc_clients.insert(std::make_pair(host, client));
            }

            void delClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.erase(host);
            }
            BaseClient::ptr getClient(const string method)
            {
                BaseClient::ptr client;
                if (_enableDiscovery)
                {
                    // 1. 通过服务发现,获取服务提供者的地址信息
                    Address host;
                    bool ret = _discovery_client->serviceDiscovery(method, host);
                    if (ret == false)
                    {
                        ELOG("当前 %s 服务,没有找到服务提供者!", method.c_str());
                        return BaseClient::ptr();
                    }
                    // 2. 查看服务提供者是否已经有实例化客户端,有则直接使用,没有则创建
                    client = getClient(host);
                    if (client.get() == nullptr)
                    {
                        // 没有找到已实例化的客户端
                        client = newClient(host);
                    }
                }
                else
                {
                    client = _rpc_client;
                }
                return client;
            }

        private:
            struct AddressHash
            {
                size_t operator()(const Address &host) const
                {
                    string addr = host.first + std::to_string(host.second);
                    return std::hash<string>{}(addr);
                }
            };
            bool _enableDiscovery;
            // 先通过discovery客户端进行服务发现
            DiscoveryClient::ptr _discovery_client;
            // 然后再进行调用
            Requestor::ptr _requestor;
            RpcCaller::ptr _caller;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _rpc_client; // 用于未启用服务发现
            std::mutex _mutex;
            //<"127.0.0.1:8080", client1>
            std::unordered_map<Address, BaseClient::ptr, AddressHash> _rpc_clients; // 用于服务发现后的连接池
        };

        // 发布订阅功能的客户端封装
        class TopicClient
        {
        public:
            using ptr = std::shared_ptr<TopicClient>;
            // 构造函数传入注册中心的地址信息,用于发送连接注册服务请求
            TopicClient(const string &ip, int port) : _requestor(std::make_shared<Requestor>()),
                                                      _topic_manager(std::make_shared<client::TopicManager>(_requestor)),
                                                      _dispatcher(std::make_shared<Dispatcher>())
            {
                // 针对topic请求后的响应进行的回调处理
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MType::RSP_TOPIC, rsp_cb);

                auto msg_cb = std::bind(&TopicManager::onPublish, _topic_manager.get(), std::placeholders::_1, std::placeholders::_2);
                // 对于主题操作请求, 使用msg_cb进行回调处理
                _dispatcher->registerHandler<TopicRequest>(MType::REQ_TOPIC, msg_cb);

                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                _rpc_client = ClientFactory::create(ip, port);
                _rpc_client->setMessageCallback(message_cb);

                // 连接中转服务器
                _rpc_client->connect();
            }
            // 向外提供的5个操作接口: 主题创建/ 删除/ 订阅/ 取消订阅/ 消息发布
            //  主题创建
            bool create(const string &key)
            {
                return _topic_manager->create(_rpc_client->connection(), key);
            }
            // 主题删除
            bool remove(const string &key)
            {
                return _topic_manager->remove(_rpc_client->connection(), key);
            }
            // 主题订阅
            bool subscribe(const string &key, const TopicManager::SubCallback &cb)
            {
                return _topic_manager->subscribe(_rpc_client->connection(), key, cb);
            }
            // 取消订阅
            bool cancel(const string &key)
            {
                return _topic_manager->cancel(_rpc_client->connection(), key);
            }
            // 消息发布
            bool publish(const string &key, const string &msg)
            {
                return _topic_manager->publish(_rpc_client->connection(), key, msg);
            }

            void shutdown()
            {
                _rpc_client->shutdown();
            }

        private:
            // 用于底层网络通信的对象
            Requestor::ptr _requestor;
            TopicManager::ptr _topic_manager; // 注意声明顺序: _topic_manager需要借助_requestor构造
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _rpc_client; // 客户端连接中转服务器
        };
    }
}